package t20240902_udf;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 3.3 连接器属性
 属性	是否必填	默认值	类型 描述
 connector	required	(none)	String	‘datagen’
 rows-per-second	optional	10000 1000010000	Long	数据生产速率
 number-of-rows	optional	(none)	Long	指定生产的数据条数，默认是不限制
 fields.#.kind	optional	random	String	指定字段的生产数据的方式 random 还是 sequence
 fields.#.min	optional	(Minimum value of type)	(Type of field)	random 生成器的指定字段 # 最小值，支持数字类型
 fields.#.max	optional	(Maximum value of type)	(Type of field)	random 生成器的指定字段 # 最大值，支持数字类型
 fields.#.length	optional	100 100100	Integer	char / varchar / string / array / map / multiset 类型的长度
 fields.#.start	optional	(none)	(Type of field)	sequence 生成器的开始值
 fields.#.end	optional	(none)	(Type of field)	sequence 生成器的结束值
 */
public class D14_UDF {

    public static void main(String[] args) {

        ParseIpPro parseIpPro = new ParseIpPro();
        System.out.println(parseIpPro.toString());

        System.setProperty("HADOOP_HOME", "D:\\software\\hadoop-3.3.6");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // String useUdf = "CREATE FUNCTION my_func AS 't20240902_udf.MyFunc'";
        String useUdf = "CREATE FUNCTION parse_ip_pro AS 't20240902_udf.ParseIpPro'";

        String genSql = "CREATE TABLE ods_tb ( " +
            " stime STRING," +
            " ip STRING," +
            " val BIGINT," +
            " rowtime AS to_timestamp(stime)," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            "  'connector' = 'kafka'," +
            "  'topic' = 'test'," +
            "  'properties.bootstrap.servers' = 'wsl:9092'," +
            "  'properties.group.id' = 'testGroup'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'" +
            ")";

        String sink = "CREATE TABLE pprint (" +
                "    stime  TIMESTAMP(3)," +
                "    ip STRING, " +
                "    val BIGINT " +
                ") WITH (" +
                "    'connector' = 'print'" +
                ")";
        String sinkSql = "INSERT INTO pprint " +
                "SELECT " +
                "rowtime, parse_ip_pro(ip), val " +
                " FROM ods_tb";

        tableEnv.executeSql(useUdf);
        tableEnv.executeSql(genSql);
        tableEnv.executeSql(sink);
        tableEnv.executeSql(sinkSql);



    }
}
